package com.hudson.admin.client;

import com.hudson.admin.config.NameSrvConfig;
import com.hudson.codec.MessageCodecFactory;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.domain.meta.TopicQueueMap;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.json.JSONUtil;
import com.hudson.core.util.StringUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

@Component
public class BrokerClient implements InitializingBean {
    private static final Logger logger = LoggerFactory.getLogger(BrokerClient.class);

    private static Bootstrap brokerClient;
    @Resource
    private NameSrvConfig config;

    public static Channel connect(String address) {
        if (StringUtil.isEmpty(address)) {
            logger.error("Broker 地址无效");
            throw new RuntimeException("Broker 地址无效");
        }
        String[] split = address.split(":");
        if (split.length == 2) {
            try {
                return brokerClient.connect(new InetSocketAddress(split[0], Integer.parseInt(split[1]))).sync().channel();
            } catch (InterruptedException e) {
                logger.error("Admin 链接 Broker 失败");
                throw new RuntimeException("Admin 链接 Broker 失败");
            }
        } else {
            logger.error("Broker 地址无效");
            throw new RuntimeException("Broker 地址无效");
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        brokerClient = new Bootstrap()
                .group(new NioEventLoopGroup(1))
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel sc) throws Exception {
                        sc.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 8, 4, 0, 0));
                        sc.pipeline().addLast(MessageCodecFactory.getCodec(config.getSerializableMethod()));
                        sc.pipeline().addLast(new IdleStateHandler(0, 10, 0, TimeUnit.SECONDS)); // 写空闲4秒触发 WRITE_IDLE
                        sc.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
                                if (evt instanceof IdleStateEvent) {
                                    IdleStateEvent event = (IdleStateEvent) evt;
                                    if (event.state() == IdleState.WRITER_IDLE) {
                                        // 发送心跳包
                                        logger.debug("produce 空闲，发送心跳至 broker...");
                                        ctx.writeAndFlush(RemotingCommand.builder().type(MessageTypeEnum.IDLE_MESSAGE.getType()).build());// 或者发送自定义心跳消息
                                    }
                                }
                            }
                        });
                        sc.pipeline().addLast(new SimpleChannelInboundHandler<RemotingCommand>() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext channelHandlerContext, RemotingCommand command) throws Exception {
                                if (MessageTypeEnum.QUEUE_MAP.equals(MessageTypeEnum.fromType(command.getType()))) {
                                    TopicQueueMap topicQueueMap = JSONUtil.fromJson(new String(command.getBody(), StandardCharsets.UTF_8), TopicQueueMap.class);
                                    BrokerClientManege.topicQueueMap.get(topicQueueMap.getBrokerName()).complete(topicQueueMap.getQueueNumMap());
                                }
                            }
                        });
                    }
                });
    }
}
